package com.example.toy.core.utils;

import com.example.toy.core.conf.ToyTableConfig;
import java.util.List;
import lombok.SneakyThrows;
import org.apache.avro.Schema;
import org.apache.avro.Schema.Field;
import org.apache.avro.Schema.Type;
import org.apache.commons.collections.ListUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * TableMetaHelper
 *
 * @author shenbing
 * @date 2021/12/21
 */
public class TableMetaHelper {

  /**
   * 判断当前文件路径是否是toy表
   * @param path
   * @return return true only when exists .toy directory under path
   */
  @SneakyThrows
  public static boolean isTablePath(Path path) {
    FileSystem fs = path.getFileSystem(new Configuration());
    Path metaDir = new Path(path, ".toy");
    return fs.getFileStatus(path).isDirectory() && fs.exists(metaDir);
  }

  /**
   * 判断两张表的配置是同一个
   * @param existedConfig
   * @param config
   */
  public static boolean isSameTable(ToyTableConfig existedConfig, ToyTableConfig config) {
    return existedConfig.getTableName().equals(config.getTableName())
        && ListUtils.isEqualList(existedConfig.getPartitionFields(), config.getPartitionFields())
        && ListUtils.isEqualList(existedConfig.getRecordKeyFields(), config.getRecordKeyFields())
        && isSameSchema(existedConfig.getSchema(), config.getSchema());
  }

  public static boolean isSameSchema(Schema s1, Schema s2) {
    if(s1.getType() != s2.getType()){
      return false;
    }
    if(s1.getType() == Type.RECORD){
      List<Field> f1 = s1.getFields();
      List<Field> f2 = s2.getFields();
      return ListUtils.isEqualList(f1, f2);
    } else if(s1.getType() == Type.UNION){
      return ListUtils.isEqualList(s1.getTypes(), s2.getTypes());
    } else {
      return true;
    }
  }
}
